package com.zhang.gmall.app

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONObject}
import com.zhang.gmall
import com.zhang.gmall.bean.{OrderDetail, OrderInfo, OrderWide}
import com.zhang.gmall.util.{JedisUtil, MyEsUtils, MyKafkaUtils, MyOffsetUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import java.time.{LocalDate, Period}
import java.util
import scala.collection.mutable.ListBuffer

/**
 * @title: 订单宽表
 * @author: zhang
 * @date: 2022/3/24 19:06 
 */
object DWDOrderApp {

  def main(args: Array[String]): Unit = {
    //TODO 1.准备实时环境
    val sparkConf: SparkConf = new SparkConf().setAppName("DWDOrderApp").setMaster("local[4]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    //TODO 2.从redis中读取offset
    val orderInfoTopicName = "DWD_ORDER_INFO_I_SPARK";
    val orderInfoGroup = "DWD_ORDER_INFO:GROUP";
    val orderInfoOffsets: Map[TopicPartition, Long] = MyOffsetUtil.getOffset(orderInfoTopicName, orderInfoGroup)
    val orderDetailTopicName = "DWD_ORDER_DETAIL_I_SPARK";
    val orderDetailGroup = "DWD_ORDER_DETAIL:GROUP";
    val orderDetailOffsets: Map[TopicPartition, Long] = MyOffsetUtil.getOffset(orderDetailTopicName, orderDetailGroup)

    //TODO 3.从kafka中消费数据
    //order_info
    var orderInfoKafkaDS: InputDStream[ConsumerRecord[String, String]] = null
    if (orderInfoOffsets != null && orderDetailOffsets.nonEmpty) {
      orderInfoKafkaDS = MyKafkaUtils.getKafkaDStream(ssc, orderInfoTopicName, orderInfoGroup, orderInfoOffsets)
    } else {
      orderInfoKafkaDS = MyKafkaUtils.getKafkaDStream(ssc, orderInfoTopicName, orderInfoGroup)
    }
    //order_detail
    var orderDetailKafkaDS: InputDStream[ConsumerRecord[String, String]] = null
    if (orderDetailOffsets != null && orderDetailOffsets.nonEmpty) {
      orderDetailKafkaDS = MyKafkaUtils.getKafkaDStream(ssc, orderDetailTopicName, orderDetailGroup, orderDetailOffsets)
    } else {
      orderDetailKafkaDS = MyKafkaUtils.getKafkaDStream(ssc, orderDetailTopicName, orderDetailGroup)
    }
    //TODO 4.提取offsets 每消费一批数据记录最后偏移量
    var orderInfoOffsetRanges: Array[OffsetRange] = null
    val orderInfoOffsetDS: DStream[ConsumerRecord[String, String]] = orderInfoKafkaDS.transform(
      rdd => {
        orderInfoOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )
    var orderDetailOffsetRanges: Array[OffsetRange] = null
    val orderDetailOffsetDS: DStream[ConsumerRecord[String, String]] = orderDetailKafkaDS.transform(
      rdd => {
        orderDetailOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )

    //TODO 5.转换结构
    val orderInfoDS: DStream[OrderInfo] = orderInfoOffsetDS.map(
      data => {
        JSON.parseObject(data.value(), classOf[OrderInfo])
      }
    )

    val orderDetailDS: DStream[OrderDetail] = orderDetailOffsetDS.map(
      data => {
        JSON.parseObject(data.value(), classOf[OrderDetail])
      }
    )

    //TODO 6.维度关联
    //5.2 维度关联
    // order_info
    val orderInfoDimDStream: DStream[OrderInfo] = orderInfoDS.mapPartitions(
      orderInfoIter => {
        //val orderInfoes: ListBuffer[OrderInfo] = ListBuffer[OrderInfo]()
        val orderInfos: List[OrderInfo] = orderInfoIter.toList
        val jedis: Jedis = JedisUtil.getJedis()
        for (orderInfo <- orderInfos) {
          //关联用户维度
          val uid: Long = orderInfo.user_id
          val redisUserKey: String = s"DIM:USER_INFO:$uid"
          val userInfoJson: String = jedis.get(redisUserKey)
          val userInfoJsonObj: JSONObject = JSON.parseObject(userInfoJson)
          //提取性别
          val gender: String = userInfoJsonObj.getString("gender")
          //提取生日
          val birthday: String = userInfoJsonObj.getString("birthday")
          //换算年龄
          val birthdayLd: LocalDate = LocalDate.parse(birthday)
          val nowLd: LocalDate = LocalDate.now()
          val period: Period = Period.between(birthdayLd, nowLd)
          val age: Int = period.getYears

          //补充到对象中
          orderInfo.user_gender = gender
          orderInfo.user_age = age


          //关联地区维度
          val provinceID: Long = orderInfo.province_id
          val redisProvinceKey: String = s"DIM:BASE_PROVINCE:$provinceID"
          val provinceJson: String = jedis.get(redisProvinceKey)
          val provinceJsonObj: JSONObject = JSON.parseObject(provinceJson)

          val provinceName: String = provinceJsonObj.getString("name")
          val provinceAreaCode: String = provinceJsonObj.getString("area_code")
          val province3166: String = provinceJsonObj.getString("iso_3166_2")
          val provinceIsoCode: String = provinceJsonObj.getString("iso_code")

          //补充到对象中
          orderInfo.province_name = provinceName
          orderInfo.province_area_code = provinceAreaCode
          orderInfo.province_3166_2_code = province3166
          orderInfo.province_iso_code = provinceIsoCode

          //处理日期字段
          val createTime: String = orderInfo.create_time
          val createDtHr: Array[String] = createTime.split(" ")
          val createDate: String = createDtHr(0)
          val createHr: String = createDtHr(1).split(":")(0)
          //补充到对象中
          orderInfo.create_date = createDate
          orderInfo.create_hour = createHr

          //orderInfoes.append(orderInfo)
        }
        jedis.close()
        orderInfos.iterator
      }
    )
    //TODO 7.转换结构为join作准备
    val orderInfoKVDS: DStream[(Long, OrderInfo)] = orderInfoDimDStream.map(orderInfo => (orderInfo.id, orderInfo))
    val orderDetailKVDS: DStream[(Long, OrderDetail)] = orderDetailDS.map(orderDetail => (orderDetail.order_id, orderDetail))

    //TODO 8.进行full join，保证join成功或者没有join成功的数据都能出现到结果集，让双方都多两部操作，到缓存中找并把自己写入到缓存
    val orderJoinDS: DStream[(Long, (Option[OrderInfo], Option[OrderDetail]))] = orderInfoKVDS.fullOuterJoin(orderDetailKVDS)

    val orderWideDS: DStream[OrderWide] = orderJoinDS.mapPartitions(
      orderJoinIter => {
        val jedis: Jedis = JedisUtil.getJedis()
        val orderWides: ListBuffer[OrderWide] = ListBuffer[OrderWide]()
        for ((key, (orderInfoOP, orderDetailOP)) <- orderJoinIter) {
          //orderInfo有，orderDetail有
          if (orderInfoOP.isDefined) {
            val orderInfo: OrderInfo = orderInfoOP.get
            if (orderDetailOP.isDefined) {
              //orderInfo有，orderDetail有
              val orderDetail: OrderDetail = orderDetailOP.get
              //组装orderWide
              val orderWide = new OrderWide(orderInfo, orderDetail)
              orderWides.append(orderWide)
            }
            //orderInfo有，orderDetail没有 （不管是否匹配成功，orderInfo都要写入缓存等待，因为一个orderInfo要与多个orderDetail关联
            //orderInfo要写入缓存等待orderDetail到来
            val orderInfoRedisKey = s"ORDERJOIN:ORDER_INFO:${orderInfo.id}"
            jedis.setex(orderInfoRedisKey, 24 * 3600, JSON.toJSONString(orderInfo, new SerializeConfig(true)))
            //同时如果orderDetail先到要去缓存中读取做关联
            val orderDetailRedisKey = s"ORDERJOIN:ORDER_DETAIL:${orderInfo.id}"
            val orderDetailS: util.Set[String] = jedis.smembers(orderDetailRedisKey)
            if (orderDetailS != null && orderDetailS.size() > 0) {
              import scala.collection.JavaConverters._
              for (elem <- orderDetailS.asScala) {
                val orderDetail: OrderDetail = JSON.parseObject(elem, classOf[OrderDetail])
                orderWides.append(new OrderWide(orderInfo, orderDetail))
              }
            }
          } else {
            //orderInfo没有，orderDetail有
            //orderDetail要去读缓存含是否有orderInfo提前到达
            val orderDetail: OrderDetail = orderDetailOP.get
            val orderInfoRedisKey = s"ORDERJOIN:ORDER_INFO:${orderDetail.order_id}"
            val orderInfoStr: String = jedis.get(orderInfoRedisKey)
            if (orderInfoStr != null && orderInfoStr.nonEmpty) {
              val orderInfo: OrderInfo = JSON.parseObject(orderInfoStr, classOf[OrderInfo])
              orderWides.append(new OrderWide(orderInfo, orderDetail))
            } else {
              //如果缓存结果中没有，则把自己加入到缓存中等待orderInfo
              val orderDetailRedisKey = s"ORDERJOIN:ORDER_DETAIL:${orderDetail.order_id}"
              jedis.set(orderDetailRedisKey, JSON.toJSONString(orderDetail, new SerializeConfig(true)))
              jedis.expire(orderDetailRedisKey, 24 * 3600)
            }
          }
        }
        jedis.close()
        orderWides.iterator
      }
    )
    //orderWideDS.print(100)
    //TODO 9.写入olap es
    orderWideDS.foreachRDD(
      rdd => {
        rdd.foreachPartition(
          orderWideIter => {
            val orderWides: List[(String, OrderWide)] = orderWideIter.map(orderWide => (orderWide.detail_id.toString, orderWide)).toList
            if (orderWides.size > 0) {
              val head: (String, OrderWide) = orderWides.head
              val date: String = head._2.create_date
              //索引名
              val indexName : String = s"gmall_order_wide_$date"
              //写入到ES
              MyEsUtils.bulkSaveIdempotent(orderWides,indexName)
            }
          }
        )
        //提交offset
        MyOffsetUtil.saveOffset(orderInfoTopicName,orderInfoGroup,orderInfoOffsetRanges)
        MyOffsetUtil.saveOffset(orderDetailTopicName,orderDetailGroup,orderDetailOffsetRanges)
      }
    )


    //TODO 10.启动任务
    ssc.start()
    ssc.awaitTermination()
  }
}
